/**
 * Copyright 2011-2011 - Members of the Chaupal Community
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

package net.chaupal.Impl.Callback;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A service to delegate task executions to a {@link ExecutorService} and
 * have it invoke a call back method when finished.
 */
public final class CallbackService {
    
    /**
     * Logger
     */
    private static final Logger LOG
            = LoggerFactory.getLogger(CallbackService.class);

    // Attributes
    
    private final ExecutorService executorService;
    private boolean shutdown = false;
    
    // Constructors
    
    public CallbackService(ExecutorService es) {
        
        // Sanity
        if ( es == null ) throw new NullPointerException();
        
        // Initialization
        this.executorService = es;
        
    }
    
    public CallbackService() {
        
        // Initialization
        this.executorService = Executors.newCachedThreadPool();
        
    }
    
    /**
     * Provides an access to the executor service.
     * 
     * @return the executor service
     */
    public final ExecutorService getExecutorService() {
        return this.executorService;
    }
    
    /**
     * Blocks until all tasks have completed execution after a shutdown
     * request, or the timeout occurs, or the current thread is
     * interrupted, whichever happens first.
     *
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @return <tt>true</tt> if this executor terminated and
     *         <tt>false</tt> if the timeout elapsed before termination
     * @throws InterruptedException if interrupted while waiting
     */    
    public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
        
        return this.executorService.awaitTermination(timeout, unit);
        
    }

    /**
     * Initiates an orderly shutdown in which previously submitted
     * tasks are executed, but no new tasks will be accepted.
     * Invocation has no additional effect if already shut down.
     *
     * @throws SecurityException if a security manager exists and
     *         shutting down this ExecutorService may manipulate
     *         threads that the caller is not permitted to modify
     *         because it does not hold {@link
     *         java.lang.RuntimePermission}<tt>("modifyThread")</tt>,
     *         or the security manager's <tt>checkAccess</tt> method
     *         denies access.
     */
    public final void shutdown() {
        this.shutdown = true;
        this.executorService.shutdown();
    }

    /**
     * Returns <tt>true</tt> if this service has been shut down.
     *
     * @return <tt>true</tt> if this service has been shut down
     */
    public final boolean isShutdown() {
        return this.shutdown;
    }
          
    /**
     * Returns <tt>true</tt> if all tasks have completed following shut down.
     * Note that <tt>isTerminated</tt> is never <tt>true</tt> unless
     * either <tt>shutdown</tt> or <tt>shutdownNow</tt> was called first.
     *
     * @return <tt>true</tt> if all tasks have completed following shut down
     */
    public final boolean isTerminated() {
        return this.executorService.isTerminated();
    }

    /**
     * Submits a callable task for execution and invokes the provided callback
     * when proceeded.
     *
     * @param <T> the return type of the callable
     * @param task the task to submit
     * @param cb a callback method
     * @return the task's future
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     */
    public final <T> Future<T> submit(Callable<T> task, CallableCallback<T> cb) {
        
        if ( this.shutdown )
            throw new RejectedExecutionException();

        // Creating callback task
        CallBackTask<T> cbt = new CallBackTask<T>(task, cb);
        
        // Returning task's future
        return cbt.proceed();
        
    }

    /**
     * Submits a callable task for execution and invokes the provided callback
     * when proceeded or if the timeout has expired.
     *
     * @param <T> the return type of the callable
     * @param task the task to submit
     * @param cb a callback method
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @return the task's future
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     */
    public final <T> Future<T> submit(Callable<T> task, CallableCallback<T> cb,
        long timeout, TimeUnit unit) {
        
        if ( this.shutdown )
            throw new RejectedExecutionException();

        // Creating callback task
        CallBackTask<T> cbt = new CallBackTask<T>(task, cb, timeout, unit);
        
        // Returning task's future
        return cbt.proceed();
        
    }
    
    private final class CallBackTask<T> implements Runnable {

        private final Callable<T> callableTask;
        private final CallableCallback<T> callableCallback;

        private final boolean withTimeout;
        
        private final long timeout;
        private final TimeUnit unit;
        
        private ExecutionException executionException = null;
        private boolean timedout = false;
        
        private Future<T> f = null;
        
        public CallBackTask(Callable<T> task, CallableCallback<T> callback,
            long timeout, TimeUnit unit) {
            
            // Sanity
            if ( task == null ) throw new NullPointerException();
            if ( callback == null ) throw new NullPointerException();
            if ( unit == null ) throw new NullPointerException();
            
            // Initialization
            this.withTimeout = true;
            this.callableTask = task;
            this.callableCallback = callback;
            this.timeout = timeout;
            this.unit = unit;
            
        }
        
        public CallBackTask(Callable<T> task, CallableCallback<T> callback) {
            
            // Sanity
            if ( task == null ) throw new NullPointerException();
            if ( callback == null ) throw new NullPointerException();
            
            // Initialization
            this.withTimeout = false;
            this.callableTask = task;
            this.callableCallback = callback;
            this.timeout = 0;
            this.unit = null;
            
        }
        
        private Future<T> proceed() {
            
            // Submitting task and this object
            f = executorService.submit(this.callableTask);
            executorService.submit(this);
            
            // Returning future
            return f;
            
        }

        @Override
        public void run() {
            
            T result = null;

            try {
            
                if ( this.withTimeout ) {
                    result = f.get(timeout, unit);
                } else {
                    result = f.get();
                }
                
            } catch (TimeoutException ex) {
                this.timedout = true;
            } catch (InterruptedException ex) {
                // This will never happen
                LOG.error("Unreachable code reached", ex);
            } catch (ExecutionException ex) {
                this.executionException = ex;
            }
            
            final T finalResult = result;
            
            CallbackEvent<T> cce = new CallbackEvent<T>() {

                @Override
                public final Callable<T> getSourceCallable() {
                    return callableTask;
                }

                @Override
                public final T getResult() {
                    return finalResult;
                }

                @Override
                public final boolean timedOut() {
                    return timedout;
                }

                @Override
                public final boolean hasExecutionException() {
                    return (executionException != null);
                }

                @Override
                public Exception getExecutionException() {
                    return executionException;
                }

                @Override
                public boolean wasCancelled() {
                    return f.isCancelled();
                }

            };

            // Calling back
            this.callableCallback.proceeded(cce);
            
        }
        
    }
    
}